package com.yd.spark.demo;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.SerializationConfig;
import org.codehaus.jackson.map.annotate.JsonSerialize;
import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple2;

public class PressureMeasure {
	private static Logger logger = LoggerFactory.getLogger(PressureMeasure.class);
	private static final ObjectMapper objectMapper = new ObjectMapper();
	static {
		objectMapper.setSerializationInclusion(JsonSerialize.Inclusion.NON_NULL); 
		objectMapper.configure(SerializationConfig.Feature.WRITE_DATES_AS_TIMESTAMPS, false);
		objectMapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
		objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);
		objectMapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
		objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true);
		objectMapper.configure(SerializationConfig.Feature.FAIL_ON_EMPTY_BEANS, false);
	}
	private static final TypeReference<Map<String, Object>> TYPE_REFERENCE = new TypeReference<Map<String, Object>>(){};
	
	public static void main(String[] args) {
		SparkConf conf = new SparkConf()
				.set("spark.network.timeout", "300");
		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> textFile = sc.textFile("/test/dispatch/*.txt");
		
		//测试一  rdd1：  计算文件字符数
		JavaRDD<Integer> lineLengths = textFile.map(new Function<String, Integer>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Integer call(String line) throws Exception {
				return line.length();
			}
			
		});
		
		int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Integer call(Integer arg0, Integer arg1) throws Exception {
				return arg0 + arg1;
			}
		});
		System.out.println("==========totalLength is :============" + totalLength);
		
		//测试二 ：  计算唯一imei数
		JavaRDD<String> imeiRDD = textFile.map(new Function<String, String>() {
			private static final long serialVersionUID = 1L;
			@Override
			public String call(String line) throws Exception {
				String[] fields = line.split(" ");
				String dispatch_snapshot = fields[12];
				try{
					Map<String, Object> paramMap = JsonUtils.fromJSON(dispatch_snapshot, TYPE_REFERENCE );
					String imei = (String) paramMap.get("imei");
					if(imei == null){
						imei="";
					}
					return imei;
				}catch (Exception e) {
					logger.error(dispatch_snapshot,e);
					return "";
				}
			}
		});
		long num = imeiRDD.distinct().count();
		System.out.println("\n==========total unique imei is :============" + num);
		//count distinct近似算法，有误差
		System.out.println("==========total imei countApproxDistinct is :============" + imeiRDD.countApproxDistinct(0.01));
		
		//测试三：rdd3 计算每个小时的imei数
		JavaPairRDD<String,String> hourly_imeiRDD = textFile.mapToPair(new PairFunction<String, String,String>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Tuple2<String, String> call(String line) throws Exception {
				String[] fields = line.split(" ");
				String dispatch_snapshot = fields[12];
				String date_hour = fields[0].split(":")[0];
				try{
					Map<String, Object> paramMap = JsonUtils.fromJSON(dispatch_snapshot, TYPE_REFERENCE );
					String imei = (String) paramMap.get("imei");
					if(imei == null){
						imei="";
					}
					return new Tuple2<String, String>(date_hour, imei);
				}catch (Exception e) {
					logger.error(dispatch_snapshot,e);
					return new Tuple2<String, String>(date_hour, "");
				}
			}
		});
		
		List<Tuple2<String, Integer>> resultList = hourly_imeiRDD.distinct().mapValues(new Function<String, Integer>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Integer call(String imei) throws Exception {
				return 1;
			}
		}).reduceByKey(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Integer call(Integer arg0, Integer arg1) throws Exception {
				return arg0 + arg1;
			}
		}).sortByKey().collect();
		
		System.out.println("\n==========count distinct imei group by hour:============");
		for(Tuple2<String, Integer> _ : resultList){
			System.out.println(_._1+" : "+_._2);
		}
		
		JavaRDD<String> badRecodeRDD = textFile.filter(new Function<String, Boolean>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Boolean call(String line) throws Exception {
				return line.split(" ", -1).length != 31;
			}
		});
//		badRecodeRDD.saveAsTextFile("/test/dispatch/bad_recodes");
		long numOfBadRecodes = badRecodeRDD.count();
		System.out.println("\n============total recodes:========:"+textFile.count());
		System.out.println("\n============num of bad recodes:========:"+numOfBadRecodes);
		//DataFrame & spark sql测试
//		testDataFrame(sc, textFile);
		
		sc.close();
	}

	public static void testDataFrame(JavaSparkContext sc,JavaRDD<String> textFile) {
		SQLContext sqlContext = new SQLContext(sc);
		String schemaString = "datetime null1 service_order_id round batch flag driver_id distance dispatch_time dispatch_lat dispatch_lng dispatch_total_rate dispatch_snapshot response_time accept_status response_lat response_lng response_distance response_time_length decision_time decision_total_rate decision_result decision_failure_reason decision_msg_snapshot subtract_amount add_price_set response_snapshot is_assigned route_distance route_time_length distance_time_length";
//		sqlContext.read().json("file.json");
		List<StructField> fields = new ArrayList<StructField>();
		for (String fieldName: schemaString.split(" ")) {
		  fields.add(DataTypes.createStructField(fieldName, DataTypes.StringType, true));
		}
		StructType schema = DataTypes.createStructType(fields);

		JavaRDD<Row> rowRDD = textFile.map(
		  new Function<String, Row>() {
			private static final long serialVersionUID = 1L;
			public Row call(String record) throws Exception {
		      String[] fields = record.split(" ",31);
		      return RowFactory.create((Object[])fields);
		    }
		  });
		DataFrame dispatchDataFrame = sqlContext.createDataFrame(rowRDD, schema);
		dispatchDataFrame.registerTempTable("dispatch");
		dispatchDataFrame.printSchema();
//		dispatchDataFrame.write().partitionBy("").mode(SaveMode.Append).save("path");
		DataFrame results = sqlContext.sql("SELECT count(distinct driver_id) FROM dispatch");
//		results.show();
		results.write().format("json").save("/test/dispatch/output/count_distinct_driver_id.json");
	}
}
